package com.shujia.flink.core;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class Demo5Join {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> studentsDS = env.socketTextStream("master", 8888);
        DataStream<String> scoreDS = env.socketTextStream("master", 9999);

        /*
         * 窗口关联:只关联窗口内的数据
         */
        JoinedStreams.WithWindow<String, String, String, TimeWindow> joinDS = studentsDS
                .join(scoreDS)
                //选择左表关联字段
                .where(stu -> stu.split(",")[0])
                //选择右表的关联字段
                .equalTo(sco -> sco.split(",")[0])
                //划分窗口
                .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(5)));

        //关联止呕整理数据
        DataStream<String> resultDS = joinDS.apply((first, second) -> first + ":" + second);

        resultDS.print();
        env.execute();
    }
}
